package com.ahaha.homeworkday2

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

object CountTest {

    case class People(id: Int, sex: String, height: String)

    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setAppName(s"${CountTest.getClass.getSimpleName}")
            .setMaster("local[*]")
        val sc = new SparkContext(conf)
        countnum(sc)
        sc.stop()
    }

    def countnum(sc: SparkContext): Unit = {
        val source = Source.fromFile("data/peopleinfo.txt", "UTF-8")
        val array = source.getLines().toArray
        source.close()
        val Peoples:Array[(String,People)] = array.map(t => {
            val strings = t.split("\\s+")
            (strings(1), new People(strings(0).toInt, strings(1), strings(2)))
        })
        val peoplesRDD:RDD[(String,People)] = sc.parallelize(Peoples)
        val value = peoplesRDD.groupByKey()
        value.foreach{case (sex,people) =>{
            println(s"${sex}对应的人数有：" + people.size)
            var max = Int.MinValue
            var min = Int.MaxValue
            for (elem <- people) {
                if (elem.height.toInt>max){
                    max = elem.height.toInt
                }
                for (elem <- people) {
                    if (elem.height.toInt<min){
                        min = elem.height.toInt
                    }
                }
            }
            println(s"${sex}对应的最高身高为：" + max)
            println(s"${sex}对应的最矮身高为：" + min)
        }}









    }
}
